package com.atguigu.flink.chapter06.Practice;

import com.atguigu.flink.bean.HotItem;
import com.atguigu.flink.bean.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

/**
 * @Author lzc
 * @Date 2023/6/21 15:40
 */
public class HotItemTopN {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        env
            .readTextFile("input/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>() {
                @Override
                public UserBehavior map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new UserBehavior(
                        Long.valueOf(data[0]),
                        Long.valueOf(data[1]),
                        Long.valueOf(data[2]),
                        data[3],
                        Long.parseLong(data[4]) * 1000, // 把时间变成毫秒
                        1L
                    );
                }
                
            })
            .filter(ub -> "pv".equals(ub.getBehavior()))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((ub, ts) -> ub.getTs())
            )
            .keyBy(UserBehavior::getItemId)
            .window(TumblingEventTimeWindows.of(Time.hours(1)))
            .reduce(
                new ReduceFunction<UserBehavior>() {
                    @Override
                    public UserBehavior reduce(UserBehavior value1, UserBehavior value2) throws Exception {
                        value1.setPvCount(value1.getPvCount() + value2.getPvCount());
                        return value1;
                    }
                },
                // 输出: 点击量 商品id 窗口的结束时间
                new ProcessWindowFunction<UserBehavior, HotItem, Long, TimeWindow>() {
                    @Override
                    public void process(Long itemId,
                                        Context ctx,
                                        Iterable<UserBehavior> elements,
                                        Collector<HotItem> out) throws Exception {
                        Long clickCount = elements.iterator().next().getPvCount();
                        HotItem hotItem = new HotItem(itemId, clickCount, ctx.window().getEnd());
                        out.collect(hotItem);
                    }
                }
            )
            .keyBy(HotItem::getWEnd)  // 按照窗口结束时间进行 KeyBy
            .process(new KeyedProcessFunction<Long, HotItem, String>() {
                Map<Long, Boolean> wEndToIsFirstMap = new HashMap<>();
                Map<Long, List<HotItem>> wEndToHotItemListMap = new HashMap<>();
                
                @Override
                public void onTimer(long timestamp,
                                    OnTimerContext ctx,
                                    Collector<String> out) throws Exception {
                    Long wEnd = ctx.getCurrentKey();
                    List<HotItem> list = wEndToHotItemListMap.get(wEnd);
                    list.sort((o1, o2) -> o2.getClickCount().compareTo(o1.getClickCount()));
                    
                    String msg = "------\n";
                    for (int i = 0, count = Math.min(3, list.size()); i < count; i++) {
                        msg += list.get(i) + "\n";
                    }
                    out.collect(msg);
                }
                
                @Override
                public void processElement(HotItem value,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    Long wEnd = ctx.getCurrentKey();
                    if (!wEndToIsFirstMap.containsKey(wEnd)) { // 如果是 true 表示不是第一个, false 表示第一个
                        // 注册一个wEnd + 2s 触发的定时器
                        ctx.timerService().registerEventTimeTimer(wEnd + 2000);
                        wEndToIsFirstMap.put(wEnd, false);
                    }
                    List<HotItem> list = wEndToHotItemListMap.getOrDefault(wEnd, new ArrayList<>());
                    list.add(value);  // 把数据存入到对应 list 集合中
                    wEndToHotItemListMap.put(wEnd, list);
                }
            })
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
每隔一小时,统计最近一小时的热门商品的 top3

热门:
    点击量

1. keyBy, 开窗,统计每个商品的点击量
    按照 商品 id keyBy
    开窗
    聚合
    
    0-1h 商品1  10000
    0-1h 商品2  2000
    0-1h 商品3  1000
    0-1h 商品3  500
    ....
    1-2h 商品3  500
    ....
    
2. 开始排序, 取 top3
    按照窗口的结束时间分组,保证相同的时间的结果在一起
    
    第一条来的注册定时器, 窗口结束+2s
    
    当定时器触发的时候,就开始排序取 top3
    
 
 */